"""
Case Type   : 数据库系统
Case Name   : 1主2备事务异步时备机备份延迟
Create At   : 2021/02/23
Owner       : songjing20
Description :
    1.初始化环境
    2.查询synchronous_standby_names,synchronous_commit值,分别设置为空和off
    3.设置recovery_min_apply_delay=2min
    4.查询参数与集群同步方式
    5.等待主备一致
    6.创建模式及表，并插入数据
    7.等待10秒,备机查询
    8.等待1min，查询备机
    9.删除表
    10.备机查询
    11.设置recovery_min_apply_delay=0
    12.备机查询
    13.清理环境
Expect      :
    1.成功
    2.设置成功
    3.设置成功
    4.返回步骤1,3修改后的值,两个备机状态为Async
    5.参数为step,主备同步
    6.创建模式及表并插入数据成功
    7.数据未同步
    8.数据同步
    9.删除表成功
    10.备机不可查询到数据
    11.设置成功
    12.无法查到该表
    13.成功
History     :
    modified by haomeng at 2025.1.17:优化用例增加等待时间以解决环境性能不足问题导致查询失败
"""
import unittest
import os
import time

from yat.test import Node
from yat.test import macro
from testcase.utils.Logger import Logger
from testcase.utils.Constant import Constant
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH


Primary_SH = CommonSH('PrimaryDbUser')


@unittest.skipIf(1 == Primary_SH.get_node_num(),
                 '需主备环境,若为单机环境则不执行')
class RecoveryDelay(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info("----this is setup----")
        self.log.info(f'----{os.path.basename(__file__)} start----')
        self.db_primary_user_node = Node('PrimaryDbUser')
        self.comshpri = CommonSH('PrimaryDbUser')
        self.constant = Constant()
        self.com = Common()
        self.conf_path = os.path.join(
            macro.DB_INSTANCE_PATH, macro.DB_PG_CONFIG_NAME)
        self.tb_name = 'tb_case0002'
        self.schema_name = 'sch_case002'
        self.rootnodelist = ['Standby1Root', 'Standby2Root']

    def test_recovery_delay(self):
            text = '----step1:初始化环境 expect:成功----'
            self.log.info(text)
            self.log.info('----1.1备份postgresqlql.conf文件----')
            shell_cmd = f"cp {self.conf_path} {self.conf_path}_testbak;" \
                f"ls -al {self.conf_path}_testbak;"
            self.log.info(shell_cmd)
            result = self.db_primary_user_node.sh(shell_cmd).result()
            self.log.info(result)
            self.assertIn(f'{self.conf_path}_testbak', result, '执行失败' +
                          text)
            self.log.info('----1.2获取节点数目----')
            self.nodelist = ['Standby1DbUser', 'Standby2DbUser']
            result = self.comshpri.get_db_cluster_status('detail')
            self.log.info(result)
            self.node_num = result.count('Standby Normal') + 1
            self.comshsta = []
            self.log.info(self.node_num)
            self.assertEqual(self.node_num, 3, '执行失败' + text)
            for i in range(int(self.node_num) - 1):
                self.comshsta.append(CommonSH(self.nodelist[i]))
            self.log.info('----1.3安装ntp服务----')
            install_cmd = "yum install -y ntp && " \
                          "systemctl enable ntpd && systemctl start ntpd"
            for i in range(int(self.node_num) - 1):
                db_standby_node = Node(node=self.rootnodelist[i])
                result = db_standby_node.sh(install_cmd).result()
                self.log.info(install_cmd)
                self.log.info(result)
            self.log.info('----1.4同步集群时间----')
            current = self.db_primary_user_node.sh(
                "date \"+%m/%d/%Y %H:%M:%S\"").result()
            self.log.info(current)
            datecmd = f'date -s "{current}"'
            for i in range(int(self.node_num) - 1):
                db_standby_node = Node(node=self.rootnodelist[i])
                result = db_standby_node.sh(datecmd).result()
                self.log.info(datecmd)
                self.log.info(result)

            text = '----step2:查询synchronous_standby_names,synchronous_commit' \
                   '值,分别设置为空和off expect:设置成功----'
            self.log.info(text)
            self.result = self.com.show_param('synchronous_standby_names')
            if self.result != '':
                re = self.comshpri.execute_gsguc(
                    "reload", self.constant.GSGUC_SUCCESS_MSG,
                    "synchronous_standby_names=''")
                self.log.info(re)
                self.assertTrue(re, '执行失败' + text)
            self.result1 = self.com.show_param('synchronous_commit')
            if self.result1 != 'off':
                re = self.comshpri.execute_gsguc(
                    'set', self.constant.GSGUC_SUCCESS_MSG,
                    'synchronous_commit=off')
                self.assertTrue(re, '执行失败' + text)
                result = self.comshpri.stop_db_cluster()
                self.assertTrue(result)
                result = self.comshpri.start_db_cluster(True)
                flg = self.constant.START_SUCCESS_MSG in result or 'Degrade' \
                      in result
                self.assertTrue(flg, '执行失败' + text)

            text = '----step3:设置recovery_min_apply_delay=2min ' \
                   'expect:设置成功----'
            self.log.info(text)
            self.result2 = self.com.show_param('recovery_min_apply_delay')
            if self.result2 != '2min':
                re = self.comshpri.execute_gsguc(
                    'reload', self.constant.GSGUC_SUCCESS_MSG,
                    'recovery_min_apply_delay=2min')
                self.assertTrue(re, '执行失败' + text)

            text = '----step4:查询参数与集群同步方式 ' \
                   'expect:返回步骤1,3修改后的值,两个备机状态为Async----'
            self.log.info(text)
            self.log.info('----查询参数----')
            result = self.com.show_param('synchronous_commit')
            self.assertIn('off', result, '执行失败' + text)
            result = self.com.show_param('recovery_min_apply_delay')
            self.assertIn('2min', result, '执行失败' + text)

            self.log.info('----查询集群同步方式----')
            sql = "select * from pg_stat_replication;"
            result = self.comshpri.execut_db_sql(sql)
            self.log.info(result)
            self.assertEqual(result.count('Async'), 2, '执行失败' + text)

            text = '----step5:等待主备一致 expect:参数为step,主备同步----'
            self.log.info(text)
            for i in range(int(self.node_num) - 1):
                result = self.comshsta[i].check_data_consistency()
                self.assertTrue(result, '执行失败' + text)

            text = '----step6:创建模式及表,并插入数据 expect:创建模式及表并插' \
                   '入数据成功----'
            self.log.info(text)
            sql = f"drop schema if exists {self.schema_name};" \
                f"create schema {self.schema_name};" \
                f"drop table if exists {self.schema_name}.{self.tb_name};" \
                f"create table {self.schema_name}.{self.tb_name}" \
                f"(i int, s char(10));" \
                f"insert into {self.schema_name}.{self.tb_name} " \
                f"values(1,'test');"
            result = self.comshpri.execut_db_sql(sql)
            self.log.info(result)
            self.assertIn(self.constant.CREATE_SCHEMA_SUCCESS_MSG, result,
                          '执行失败' + text)
            self.assertIn(self.constant.TABLE_CREATE_SUCCESS, result,
                          '执行失败' + text)
            self.assertIn(self.constant.INSERT_SUCCESS_MSG, result, '执行失败'
                          + text)

            text = '----step7:等待10秒,备机查询 expect:数据未同步----'
            self.log.info(text)
            time.sleep(10)
            sql = f"select sysdate;" \
                f"select * from {self.schema_name}.{self.tb_name};"
            for i in range(int(self.node_num) - 1):
                result = self.comshsta[i].execut_db_sql(sql)
                self.log.info(result)
                self.assertIn(self.constant.NOT_EXIST, result, '执行失败' +
                              text)

            text = '----step8:等待2min,备机查询 expect:数据同步----'
            self.log.info(text)
            time.sleep(150)
            sql = f"select sysdate;" \
                f"select * from {self.schema_name}.{self.tb_name};"
            for i in range(int(self.node_num) - 1):
                result = self.comshsta[i].execut_db_sql(sql)
                self.log.info(result)
                self.assertIn('1 | test', result, '执行失败' + text)

            text = '----step9:删除表 expect:删除表成功----'
            self.log.info(text)
            sql = f"select sysdate;" \
                f"drop table if exists {self.schema_name}.{self.tb_name};"
            result = self.comshpri.execut_db_sql(sql)
            self.log.info(result)
            self.assertIn(self.constant.DROP_TABLE_SUCCESS, result, '执行失败'
                          + text)

            text = '----step10:备机查询 expect:备机不可查询到数据----'
            self.log.info(text)
            time.sleep(150)
            sql = f"select sysdate;" \
                f"select * from {self.schema_name}.{self.tb_name};"
            for i in range(int(self.node_num) - 1):
                result = self.comshsta[i].execut_db_sql(sql)
                self.log.info(result)
                self.assertIn(self.constant.NOT_EXIST, result, '执行失败' +
                              text)

            text = '----step11:设置recovery_min_apply_delay=0 ' \
                   'expect:设置成功----'
            self.log.info(text)
            result = self.comshpri.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                'recovery_min_apply_delay=0')
            self.assertTrue(result, '执行失败' + text)

            text = '----step12:查询备机 expect:无法查到该表----'
            self.log.info(text)
            sql = f"select sysdate;select * from {self.tb_name};"
            for i in range(int(self.node_num) - 1):
                result = self.comshsta[i].execut_db_sql(sql)
                self.log.info(result)
                self.assertIn(self.constant.NOT_EXIST, result, '执行失败' +
                              text)

    def tearDown(self):
        text = '----step13:清理环境 expect:成功----'
        self.log.info(text)

        self.log.info('----删除模式----')
        sql = f"drop schema if exists {self.schema_name} cascade;"
        result_schema = self.comshpri.execut_db_sql(sql)
        self.log.info(result_schema)

        self.log.info('----还原配置文件----')
        shell_cmd = f"rm -rf {self.conf_path};" \
            f"cp {self.conf_path}_testbak {self.conf_path};" \
            f"rm -rf {self.conf_path}_testbak"
        self.log.info(shell_cmd)
        result_conf = self.db_primary_user_node.sh(shell_cmd).result()
        self.log.info(result_conf)
        shell_cmd1 = f"ls -al {self.conf_path};"
        self.log.info(shell_cmd1)
        shell_cmd2 = f"ls -al {self.conf_path}_testbak;"
        self.log.info(shell_cmd2)
        result_conf1 = self.db_primary_user_node.sh(shell_cmd1).result()
        self.log.info(result_conf1)
        result_conf2 = self.db_primary_user_node.sh(shell_cmd2).result()
        self.log.info(result_conf2)

        self.log.info('----还原synchronous_standby_names,synchronous_commit,'
                      'recovery_min_apply_delay----')
        cmd = f"alter system set synchronous_standby_names to '{self.result}'"
        re1 = self.comshpri.execut_db_sql(cmd)
        self.log.info(re1)
        re2 = self.comshpri.execute_gsguc(
            "reload", self.constant.GSGUC_SUCCESS_MSG,
            f"synchronous_commit='{self.result1}'")
        re3 = self.comshpri.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            f'recovery_min_apply_delay={self.result2}')

        self.assertIn('ALTER', re1, '执行失败' + text)
        self.assertTrue(re2, '执行失败' + text)
        self.assertTrue(re3, '执行失败' + text)
        self.assertNotIn('No such file or directory', result_conf1, '执行失败'
                         + text)
        self.assertIn('No such file or directory', result_conf2, '执行失败' +
                      text)
        self.assertIn(self.constant.DROP_SCHEMA_SUCCESS_MSG, result_schema,
                      '执行失败' + text)
        self.log.info(f'----{os.path.basename(__file__)} end----')
